/**
 * 
 */
package demo.java.lang.management.jvm;

import demo.dcs.DistinctServiceIdHolder;
import demo.dcs.vo.OnlineStatus;
import demo.java.lang.management.vo.ClientConstants;
import demo.java.lang.management.vo.EventType;
import demo.java.lang.management.vo.JsonUtils;
import demo.java.lang.management.vo.MetricSpan;
import demo.java.lang.management.vo.SpanType;
import demo.java.lang.management.vo.XdcsLogger;
import demo.java.lang.management.vo.XdcsLoggerUtil;
import demo.java.lang.management.vo.XmlyLogSpan;
import demo.java.lang.management.vo.XmlySpan;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;


/**
 * @author mandy.hu
 *
 */

public class LocalAsySendImpl2 {

	private static Logger logger = LoggerFactory.getLogger(LocalAsySendImpl2.class);

	private ArrayBlockingQueue<XmlySpan> monitorQueue;
	
	private ArrayBlockingQueue<XmlySpan> queue;
	
	private ArrayBlockingQueue<XmlyLogSpan> logQueue;

    private DistinctServiceIdHolder distinctServiceIdHolder = DistinctServiceIdHolder.getInstance();

	private List<XmlySpan> spansCache;
	
	private List<XmlyLogSpan> logsCache;

	private List<XmlySpan> monitorsCache;

	private volatile boolean isReady = false;
	
	private TransferTask task;
	
	private TransferTask4Log task4Log;
	
	private TransferTask4Monitor task4Monitor;

	private int writeQueueSize = AppTraceConfig.getLocalConfig().getWriteQueueSize();
	

	static{
		XdcsLoggerUtil.dealWithLogger(AppTraceConfig.getLocalConfig().getPersonalUrlMapping());
		XdcsLoggerUtil.dealWithLogger(AppTraceConfig.getLocalConfig().getUrl2eventMapping());
		XdcsLoggerUtil.dealWithLogger(AppTraceConfig.getLocalConfig().getUrlContain2eventMapping());
		XdcsLoggerUtil.getLoggerByName("log", AppTraceConfig.getLocalConfig().getLogDir());
		XdcsLoggerUtil.getLoggerByName("monitor", AppTraceConfig.getLocalConfig().getLogDir());
		XdcsLoggerUtil.getLoggerByName("trace", AppTraceConfig.getLocalConfig().getLogDir());
		XdcsLoggerUtil.getLoggerByName("metric", AppTraceConfig.getLocalConfig().getLogDir());
		XdcsLoggerUtil.getLoggerByName("jvmStatic", AppTraceConfig.getLocalConfig().getLogDir());
		XdcsLoggerUtil.getLoggerByName("jvmDynamic", AppTraceConfig.getLocalConfig().getLogDir());
		XdcsLoggerUtil.getLoggerByName("statsample", AppTraceConfig.getLocalConfig().getLogDir());
		XdcsLoggerUtil.getLoggerByName("dynamiclog", AppTraceConfig.getLocalConfig().getLogDir());

		try{
//			XdcsHystrixConcurrencyStrategy.initRegister();
			logger.warn("XdcsHystrixConcurrencyStrategy.initRegister success");
		}catch (Exception e){
			logger.warn("XdcsHystrixConcurrencyStrategy.initRegister",e.getMessage());
		}
	}
	
	public LocalAsySendImpl2() {
		if(writeQueueSize == 0){
			writeQueueSize = 4096;
		}
		this.queue = new ArrayBlockingQueue<XmlySpan>(writeQueueSize);
		this.monitorQueue = new ArrayBlockingQueue<XmlySpan>(writeQueueSize);
        this.monitorsCache = new ArrayList<XmlySpan>();
        this.spansCache = new ArrayList<XmlySpan>();

        this.logQueue = new ArrayBlockingQueue<XmlyLogSpan>(512);
        this.logsCache = new ArrayList<XmlyLogSpan>();

		this.task = new TransferTask();
		this.task4Log = new TransferTask4Log();
		this.task4Monitor = new TransferTask4Monitor();
		this.start();
	}

	public LocalAsySendImpl2(int spanSzie) {
		this.queue = new ArrayBlockingQueue<XmlySpan>(spanSzie);
		this.monitorQueue = new ArrayBlockingQueue<XmlySpan>(spanSzie);
        this.monitorsCache = new ArrayList<XmlySpan>();
		this.spansCache = new ArrayList<XmlySpan>();
		
		this.logQueue = new ArrayBlockingQueue<XmlyLogSpan>(512);
		this.logsCache = new ArrayList<XmlyLogSpan>();

		this.task = new TransferTask();
		this.task4Log = new TransferTask4Log();
		this.task4Monitor = new TransferTask4Monitor();
		this.start();
	}


	private class TransferTask4Log extends Thread{
		TransferTask4Log() {
			this.setName("TransferTask4Log-agent-Thread");
		}

		@Override
		public void run() {
			for (;;) {
				try {
					if (!isReady()) {
                        isReady = true;
                        } else {
						while (!task4Log.isInterrupted()) {
                            logsCache.clear();
                            XmlyLogSpan logSpan = logQueue.take();
                            logsCache.add(logSpan);
                            logQueue.drainTo(logsCache);
                            this.conSendLogs(logsCache);
						}
					}
				} catch (Throwable e) {
                    logger.error("Transfer xmlyLogSpan error",e);
				}
			}
		}

		private void conSendLogs(List<XmlyLogSpan> logsCache) {
			for (XmlyLogSpan xmlyLogSpan : logsCache) {
				String type = xmlyLogSpan.getProps().get(ClientConstants.SPAN_TYPE);
				String eventType = xmlyLogSpan.getEventType();
				try {
					if(StringUtils.equals(EventType.SOUTHGATE_ERRORLOG.getEventType(),eventType)){
						XdcsLoggerUtil.getLoggerByName("southgateErrorLog",
								AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(xmlyLogSpan));
						continue;
					}

					if(StringUtils.equals(ClientConstants.DYNAMICLOG,type)){
						XdcsLoggerUtil.getLoggerByName("dynamiclog", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(xmlyLogSpan));
					}else{
						XdcsLoggerUtil.getLoggerByName("log", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(xmlyLogSpan));
					}
				}catch (Throwable e){
	                logger.error("Transfer conSendLogs error::",e);
	            }
			}
            
        }
		
	}

	private class TransferTask4Monitor extends Thread{
		TransferTask4Monitor() {
			this.setName("TransferTask4Monitor-agent-Thread");
		}

		@Override
		public void run() {
			for (;;) {
				try {
					if (!isReady()) {
                        isReady = true;
                        } else {
						while (!task4Monitor.isInterrupted()) {
							monitorsCache.clear();
                            XmlySpan logSpan = monitorQueue.take();
                            monitorsCache.add(logSpan);
                            monitorQueue.drainTo(monitorsCache);
                            this.conSendMonitors2(monitorsCache);
						}
					}
				} catch (Throwable e) {
                    logger.error("Transfer xmlyMonitorSpan error",e);
				}
			}
		}
		private void conSendMonitors2(List<XmlySpan> monitorsCache) {
			if (monitorsCache!=null && monitorsCache.size()>0) {
				List<XmlySpan> dbMonitor =  new ArrayList<XmlySpan>();
				List<XmlySpan> serverMonitor =  new ArrayList<XmlySpan>();
				List<XmlySpan> clientMonitor =  new ArrayList<XmlySpan>();
				for (XmlySpan xmlySpan : monitorsCache) {
					if (StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(), SpanType.MAINSTAY_CLIENT.getSpanType()) || StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(),
							SpanType.THRIFTCLIENT.getSpanType()) || StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(),SpanType.HTTP.getSpanType())) {
						clientMonitor.add(xmlySpan);
					}else if (StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(), SpanType.MAINSTAY_SERVER.getSpanType()) || StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(),
							SpanType.HTTP_SERVER.getSpanType()) || StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(),SpanType.THRIFTSERVER.getSpanType())) {
						serverMonitor.add(xmlySpan);
					}else if (StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(), SpanType.SQL.getSpanType()) || StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(),
							SpanType.REDIS.getSpanType()) || StringUtils.equalsIgnoreCase(xmlySpan.getSpanType().getSpanType(),SpanType.METHOD.getSpanType())) {
						dbMonitor.add(xmlySpan);
						
					}else {
						serverMonitor.add(xmlySpan);
					}
				}
				if (dbMonitor!=null && dbMonitor.size()>0) {
					XdcsLoggerUtil.getLoggerByName("storeMonitor",AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(dbMonitor));
				}
				if (serverMonitor!=null && serverMonitor.size()>0) {
					XdcsLoggerUtil.getLoggerByName("monitor",AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(serverMonitor));
				}
				if (clientMonitor!=null && clientMonitor.size()>0) {
					XdcsLoggerUtil.getLoggerByName("clientMonitor",AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(clientMonitor));
				}
			}
            
        }
		
		private void conSendMonitors(List<XmlySpan> monitorsCache) {
			if (monitorsCache!=null && monitorsCache.size()>0) {
				XdcsLoggerUtil.getLoggerByName("monitor",AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(monitorsCache));
			}
            
        }
		
	}

	private class TransferTask extends Thread {
		TransferTask() {
			this.setName("TransferTask-client-Thread");
		}

		@Override
		public void run() {
			for (;;) {
				try {
					if (!isReady()) {
                        isReady = true;
					} else {
						while (!task.isInterrupted()) {
                            spansCache.clear();
                            XmlySpan first = queue.take();
                            spansCache.add(first);
                            queue.drainTo(spansCache);
                            this.conSends(spansCache);
						}
					}
                } catch (Throwable e) {
                    logger.error("Transfer xmlySpan error",e);
				}
			}
		}
		private void conSends(List<XmlySpan> spansCache) {
			for (XmlySpan event : spansCache) {
				try {
					if (StringUtils.isNotBlank(event.getPersonalType())) {
						XdcsLoggerUtil.getLoggerByName(event.getPersonalType(), AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));

                        if (StringUtils.equals(event.getAppName(), "xdcs-collector")
								||(StringUtils.startsWith(event.getPersonalType(), "dxl")&& StringUtils.endsWith(event.getPersonalType(), "Tp"))
								|| StringUtils.equalsIgnoreCase(event.getPersonalType(), "mainstay-metric")){
                            continue;
                        }
					}
                    //else {
						if (event.getEventType()==null) {
							XdcsLoggerUtil.getLoggerByName("trace", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
							
						}else {
							switch (event.getEventType()) {
							case AD:
								if (StringUtils.equalsIgnoreCase(event.getStatus(), OnlineStatus.OFF.getStatus())) {
									XdcsLoggerUtil.getLoggerByName("offline-ad", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								}else {
									XdcsLoggerUtil.getLoggerByName("ad", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								}
								break;
							case CDN:
								XdcsLoggerUtil.getLoggerByName("cdn", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));			
								break;
							case PLAY:
								if (StringUtils.equalsIgnoreCase(event.getStatus(), OnlineStatus.OFF.getStatus())) {
									XdcsLoggerUtil.getLoggerByName("offline-play", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								}else {
									XdcsLoggerUtil.getLoggerByName("play", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								}
								break;
							case RECOMMEND:
								XdcsLoggerUtil.getLoggerByName("recommend", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							case DOWNLOAD:
								XdcsLoggerUtil.getLoggerByName("download", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							case SEARCH:
								XdcsLoggerUtil.getLoggerByName("search", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							case ERRORLOG:
								XdcsLoggerUtil.getLoggerByName("error", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							case PUSH:
								XdcsLoggerUtil.getLoggerByName("push", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							case CRASH:
								XdcsLoggerUtil.getLoggerByName("crash", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								 break;
							case CLICK2HDFS:
								XdcsLoggerUtil.getLoggerByName(EventType.CLICK2HDFS.getEventType(), AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							case COLLECTOR:
								XdcsLoggerUtil.getLoggerByName(EventType.COLLECTOR.getEventType(), AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							case MOBILE:
								XdcsLoggerUtil.getLoggerByName(EventType.MOBILE.getEventType(), AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
//				            case LIVE:
//									XdcsLoggerUtil.getLoggerByName("live", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
//									break;
							default:
								XdcsLoggerUtil.getLoggerByName("trace", AppTraceConfig.getLocalConfig().getLogDir()).info(JsonUtils.toJson(event));
								break;
							}
						}
					//}
				} catch (Throwable e) {
					logger.error("Transfer conSends error::",e);
				}
				
				
			}
            
		}

	}


	public boolean isReady() {
		return isReady;
	}

	public void start() {
		
		if (!task.isAlive()) {
			task.setName("write trace log-" + task.getName());
			task.setDaemon(true);
			task.start();
			Runtime.getRuntime().addShutdownHook(new Thread() {
				public void run() {
					cancel();
				}
			});
		}
		if (!task4Log.isAlive()) {
			task4Log.setName("write log-" + task4Log.getName());
			task4Log.setDaemon(true);
			task4Log.start();
			Runtime.getRuntime().addShutdownHook(new Thread() {
				public void run() {
					task4Log.interrupt();
				}
			});
		}
		if (!task4Monitor.isAlive()) {
			task4Monitor.setName("write Monitorlog-" + task4Monitor.getName());
			task4Monitor.setDaemon(true);
			task4Monitor.start();
			Runtime.getRuntime().addShutdownHook(new Thread() {
				public void run() {
					task4Monitor.interrupt();
				}
			});
		}
		
	}

	public void cancel() {
		task.interrupt();
	}
	
	public void asySend(XmlySpan span) {
		try {
            if (distinctServiceIdHolder.addServiceId(span)){
            	
                queue.add(span);
            }
        } catch (Throwable e) {
			logger.error("span : ignore ..");
		}
	}

    public void asyCustomMetricSends(MetricSpan metricSpan) {
        try {
            if (metricSpan!=null){
				XdcsLogger logger = XdcsLoggerUtil.getLoggerByName("customMetric", AppTraceConfig.getLocalConfig().getLogDir());
                logger.info(JsonUtils.toJson(metricSpan));
            }
        } catch (Throwable e) {
            logger.error("custom metric span : ignore ..");
        }
    }

    public void asyCustomMetricSends(List<MetricSpan> spans) {
        try {
            if (spans!=null && spans.size()>0){
                for (MetricSpan metricSpan : spans){
                    asyCustomMetricSends(metricSpan);
                }
            }
        } catch (Throwable e) {
            logger.error("custom metric span list : ignore ..");
        }
    }

	public void asyMetricSends(List<MetricSpan> spans) {
		try {
            if (spans!=null && spans.size()>0){
				XdcsLogger logger = XdcsLoggerUtil.getLoggerByName("metric", AppTraceConfig.getLocalConfig().getLogDir());
                for (MetricSpan metricSpan:spans){
                    logger.info(JsonUtils.toJson(metricSpan));
                }
            }
		} catch (Throwable e) {
            logger.error("monitor span list : ignore ..");
		}

	}

    public void jvmStaticInfoSends(String jvmStatic) {
        try {
            if (StringUtils.isNotBlank(jvmStatic)){
				XdcsLogger logger = XdcsLoggerUtil.getLoggerByName("jvmStatic", AppTraceConfig.getLocalConfig().getLogDir());
                logger.info(jvmStatic);
            }
        } catch (Throwable e) {
            logger.error("jvmStaticInfoSends info dump : ignore ..");
        }

    }

    public void jvmDynamicInfoSends(List<String> threadInfos) {
        try {
            if (threadInfos!=null && threadInfos.size()>0){
				XdcsLogger logger = XdcsLoggerUtil.getLoggerByName("jvmDynamic", AppTraceConfig.getLocalConfig().getLogDir());
                for (String threadInfo : threadInfos){
                    logger.info(threadInfo);
                }
            }
        } catch (Throwable e) {
            logger.error("jvmDynamicInfoSends info dump : ignore ..");
        }

    }

	public void asyMonitorSend(XmlySpan span) {
		try {
		    //client数据不进行落盘
		    if(StringUtils.equalsIgnoreCase(span.getSpanType().getSpanType(), SpanType.MAINSTAY_CLIENT.getSpanType())
                    || StringUtils.equalsIgnoreCase(span.getSpanType().getSpanType(), SpanType.THRIFTCLIENT.getSpanType())){
		        return;
            }

			if (distinctServiceIdHolder.tooManySCKeys()) {
				return;
			}

			if (span.getIsSample() && AppTraceConfig.getLocalConfig().isWriteMainstayLog()) {
				monitorQueue.add(span);
			}

        } catch (Throwable e) {
			logger.error("monitor span : ignore ..");
		}
	}

	public void asyMonitorSends(List<XmlySpan> spans) {
		try {
            if (spans!=null && spans.size()>0){
                for (XmlySpan xmlySpan : spans){
                    if (xmlySpan.getIsSample()) {
                        monitorQueue.add(xmlySpan);
                    }
                }
            }
            //采样补偿
            //SampleCompensation.getSampleCompensation().dealWithSpan(spans);
		} catch (Throwable e) {
            logger.error("monitor span list : ignore ..");
		}

	}
	
	public void conSend(String topic,String data) {
		XdcsLoggerUtil.getLoggerByName("trace", AppTraceConfig.getLocalConfig().getLogDir()).info(topic + "#/#" + data);
		
	}

	public void asyLandingData(String json) {
		XdcsLoggerUtil.getLoggerByName("landing", AppTraceConfig.getLocalConfig().getLogDir()).info(json);
	}

	public void asySends(List<XmlySpan> spans) {
		try {
            for (XmlySpan span : spans){
                if (distinctServiceIdHolder.addServiceId(span)){
                    queue.add(span);
                }
            }
		} catch (Throwable e) {
            logger.error("span list : ignore ..");
		}

	}

	public void asySendLogSpan(XmlyLogSpan logSpan) {
		if (distinctServiceIdHolder.tooManyServiceIds() || distinctServiceIdHolder.tooManySCKeys()) {
			return;
		}
		try {
			logQueue.add(logSpan);
            //logger.info("asy send logspan:"+logSpan);
			} catch (Throwable e) {
            logger.error("logSpan : ignore ..", e);
		}
	}

	public void asySendLogSpans(List<XmlyLogSpan> logSpans) {
		try {
			logQueue.addAll(logSpans);
		} catch (Throwable e) {
            logger.error("logSpan list : ignore ..",e);
		}

	}
	

}
